package com.zf.spring.microservice.zuul.common.filters.route;

import java.io.IOException;
import java.io.InputStream;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.servlet.http.HttpServletRequest;

import org.apache.commons.io.Charsets;
import org.apache.commons.io.IOUtils;
import org.apache.http.HttpResponse;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.netflix.ribbon.support.RibbonRequestCustomizer;
import org.springframework.cloud.netflix.zuul.filters.ProxyRequestHelper;
import org.springframework.cloud.netflix.zuul.filters.route.RibbonCommand;
import org.springframework.cloud.netflix.zuul.filters.route.RibbonCommandContext;
import org.springframework.cloud.netflix.zuul.filters.route.RibbonCommandFactory;
import org.springframework.http.client.ClientHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.util.MultiValueMap;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.functions.FuncN;
import rx.schedulers.Schedulers;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.netflix.zuul.ZuulFilter;
import com.netflix.zuul.context.RequestContext;
import com.zf.spring.microservice.zuul.common.MergeRouteInfo;
import com.zf.spring.microservice.zuul.common.MergeStepInfo;
import com.zf.spring.microservice.zuul.common.filters.FilterConfigurationBean;
import com.zf.spring.microservice.zuul.common.filters.SimpleHostRouteHelper;

/**
 * 服务聚合处理filter<br/>
 * 这个类，对携带header等信息处理的不好，需要注意这一点。<br/>
 * @see com.zf.spring.microservice.zuul.common.filters.pre.PreMergeFilter
 * @see org.springframework.cloud.netflix.zuul.filters.route.SimpleHostRoutingFilter
 * @see org.springframework.cloud.netflix.zuul.filters.route.RibbonRoutingFilter
 * @author zhangfeng
 * 
 */
@Component
public class MergeRoutingFilter extends ZuulFilter {
	Logger log = Logger.getLogger(MergeRoutingFilter.class);
	FilterConfigurationBean filterConfigurationBean;
	protected ProxyRequestHelper helper;
	protected RibbonCommandFactory<?> ribbonCommandFactory;
	@Autowired(required = false)
	protected List<RibbonRequestCustomizer> requestCustomizers = Collections
			.emptyList();
	SimpleHostRouteHelper simpleHostRouteHelper;

	public MergeRoutingFilter(FilterConfigurationBean filterConfigurationBean,
			ProxyRequestHelper helper,
			RibbonCommandFactory<?> ribbonCommandFactory,
			SimpleHostRouteHelper simpleHostRouteHelper) {
		this.filterConfigurationBean = filterConfigurationBean;
		this.helper = helper;
		this.ribbonCommandFactory = ribbonCommandFactory;
		this.simpleHostRouteHelper = simpleHostRouteHelper;
	}

	@Override
	public boolean shouldFilter() {
		RequestContext ctx = RequestContext.getCurrentContext();
		// 有此参数，代表为聚合服务
		// @see com.zf.spring.microservice.zuul.common.filters.pre.PreMergeFilter
		return ctx.get("mergeRouteId") != null;
	}

	/**
	 * 目前实现并发多个请求，并组装 下一步：有序请求 下下一步：响应结果复用
	 */
	@Override
	public Object run() {
		final java.util.concurrent.atomic.AtomicBoolean runningModel = new AtomicBoolean(
				true);
		final JSONObject responseMsg = new JSONObject();
		RequestContext ctx = RequestContext.getCurrentContext();
		// 本意是想尽可能保留一些信息.... 这里跳过，没仔细看了
		this.helper.addIgnoredHeaders();
		// 获取服务调用步骤信息
		MergeRouteInfo mergeRouteInfo = (MergeRouteInfo) ctx
				.get("mergeRouteInfo");
		// 获取开始的步骤
		List<String> starts = mergeRouteInfo.getStarts();
		final Map<String, MergeStepInfo> steps = mergeRouteInfo.getSteps();
		// 目前仅实现了第一步，也就是start....调用顺序还没实现
		// Observable -> RxJava中的概念
		// 这里的意思是，将多个请求，存在集合中，再使用zip操作符并发调用
		ArrayList<Observable<MergeStepInfo>> reqs = new ArrayList<Observable<MergeStepInfo>>();
		for (final String name : starts) {
			Observable<MergeStepInfo> req = null;
			MergeStepInfo stepInfo = steps.get(name);
			if (stepInfo.getLocation().startsWith("https")
					|| stepInfo.getLocation().startsWith("http")) {
				// http开头的使用httpclient调用方式
				req = doWithHttpClient(stepInfo);
			} else {
				// 服务ID，使用ribbon调用 
				req = doWithRibbon(stepInfo);
			}
			reqs.add(req);
		}
		// zip会在所有的步骤完成后，通知订阅的对象onNext
		Observable.zip(reqs, new FuncN<List<MergeStepInfo>>() {

			@Override
			public List<MergeStepInfo> call(Object... obj) {
				// 这里是将每一步调用返回的结果装进集合，再通知订阅者
				List<MergeStepInfo> mergeStepInfos = new ArrayList<MergeStepInfo>();
				for (Object o : obj) {
					mergeStepInfos.add((MergeStepInfo) o);
				}
				return mergeStepInfos;
			}
		}).subscribe(new Observer<List<MergeStepInfo>>() {

			@Override
			public void onCompleted() {
				runningModel.set(false);
			}

			@Override
			public void onError(Throwable e) {
				log.warn(e.getMessage(), e);
				runningModel.set(false);
			}

			@Override
			public void onNext(List<MergeStepInfo> t) {
				responseMsg.put("result", "true");
				responseMsg.put("datas", JSONArray.toJSON(t));
			}
		});
		// 因为上面的zip是异步处理，所有写了个循环在这里等待合并完结
		while (runningModel.get()) {
		}
		// 把结果塞到response里边
		ctx.setResponseGZipped(false);
		ctx.setResponseDataStream(IOUtils.toInputStream(
				responseMsg.toJSONString(), Charsets.UTF_8));
		// 因为上下文中的serviceId已经被清除，后续的routefilter不会再执行..
		return null;
	}

	// http请求
	private Observable<MergeStepInfo> doWithHttpClient(
			final MergeStepInfo stepInfo) {
		RequestContext context = RequestContext.getCurrentContext();
		final HttpServletRequest request = context.getRequest();
		final MultiValueMap<String, String> headers = this.helper
				.buildZuulRequestHeaders(request);
		final MultiValueMap<String, String> params = this.helper
				.buildZuulRequestQueryParams(request);
		// http method
		final String verb = stepInfo.getMethod();
		// 请求参数转为流
		final InputStream requestEntity = IOUtils.toInputStream(stepInfo
				.getData());
		if (stepInfo.getData() == null || stepInfo.getData().length() <= 0) {
			context.setChunkedRequestBody();
		}
		// 请求URL
		final String uri = stepInfo.getPath();
		this.helper.addIgnoredHeaders();

		try {
			return Observable.create(
					new Observable.OnSubscribe<MergeStepInfo>() {

						@Override
						public void call(Subscriber<? super MergeStepInfo> t) {
							try {
								// 执行请求...此处自定义的原因是...zuul上下文保存用的是线程变量，我新开线程以后，上下文就木有啦..很郁闷
								// 这里就是做了一个http请求，本想直接复制过来使用，结果没弄透zuul，把自己坑了
								HttpResponse response = simpleHostRouteHelper.forward(
										simpleHostRouteHelper.getHttpClient(),
										verb, uri, headers, params,
										requestEntity,
										new URL(stepInfo.getLocation()),
										stepInfo);
								stepInfo.setResult(IOUtils.toString(response
										.getEntity().getContent()));
								t.onNext(stepInfo);
							} catch (Exception e) {
								t.onError(e);
							}
							t.onCompleted();
						}
					}).subscribeOn(Schedulers.newThread());

		} catch (Exception ex) {
			throw ex;
		}
	}

	/**
	 * 每一步，都将封装为一个Observable
	 * @param stepInfo
	 * @return
	 */
	private Observable<MergeStepInfo> doWithRibbon(final MergeStepInfo stepInfo) {
		RequestContext ctx = RequestContext.getCurrentContext();
		HttpServletRequest request = ctx.getRequest();
		MultiValueMap<String, String> headers = this.helper
				.buildZuulRequestHeaders(request);

		MultiValueMap<String, String> params = this.helper
				.buildZuulRequestQueryParams(request);
		String verb = stepInfo.getMethod();
		InputStream requestEntity = IOUtils.toInputStream(stepInfo.getData());
		if (stepInfo.getData() == null || stepInfo.getData().length() < 0) {
			ctx.setChunkedRequestBody();
		}

		String serviceId = stepInfo.getLocation();
		Boolean retryable = (Boolean) ctx.get("retryable");

		String uri = stepInfo.getPath();

		// remove double slashes
		uri = uri.replace("//", "/");

		long contentLength = stepInfo.getData().length();
		// 创建command需要的上下文信息..
		RibbonCommandContext ribbonCommandContext = new RibbonCommandContext(
				serviceId, verb, uri, retryable, headers, params,
				requestEntity, this.requestCustomizers, contentLength);
		// 调用factory
		final RibbonCommand command = this.ribbonCommandFactory
				.create(ribbonCommandContext);
		try {
			return Observable.create(
					new Observable.OnSubscribe<MergeStepInfo>() {

						@Override
						public void call(Subscriber<? super MergeStepInfo> t) {

							ClientHttpResponse response = command.execute();
							try {
								stepInfo.setResult(IOUtils.toString(
										response.getBody(), Charsets.UTF_8));
								t.onNext(stepInfo);
							} catch (IOException e) {
								t.onError(e);
							}
							t.onCompleted();
						}
					}).subscribeOn(Schedulers.newThread());
		// subscribeOn(Schedulers.newThread()) 这个意思是，开启新线程处理
		} catch (Exception ex) {
			throw ex;
		}
	}

	@Override
	public String filterType() {
		return "route";
	}

	@Override
	public int filterOrder() {
		// before RibbonRoutingFilter
		return 9;
	}

}
